package com.atguigu.day08;

import com.atguigu.beans.WaterSensor;
import com.atguigu.func.WaterSensorMapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author Felix
 * @date 2024/4/9
 * 该案例演示了算子状态的广播状态
 * 需求：水位超过指定的阈值发送告警，阈值可以动态修改
 */
public class Flink09_OpeState_BroadcastState {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(2);
        //env.setStateBackend(new HashMapStateBackend());

        //env.setStateBackend(new EmbeddedRocksDBStateBackend());
        //TODO 2.从指定的网络端口读取水位数据
        SingleOutputStreamOperator<WaterSensor> wsDS = env
                .socketTextStream("hadoop102", 8888)
                .map(new WaterSensorMapFunction());

        //TODO 3.从指定的网络端口读取阈值数据
        SingleOutputStreamOperator<Integer> thresholdDS = env
                .socketTextStream("hadoop102", 8889)
                .map(Integer::parseInt);

        //TODO 4.广播阈值流  --- 定义广播状态描述器
        //声明广播状态描述器
        MapStateDescriptor<String, Integer> mapStateDescriptor
                = new MapStateDescriptor<String, Integer>("mapStateDescriptor",String.class, Integer.class);

        BroadcastStream<Integer> broadcastDS = thresholdDS.broadcast(mapStateDescriptor);

        //TODO 5.将水位流和广播流进行连接---connect
        BroadcastConnectedStream<WaterSensor, Integer> connectDS = wsDS.connect(broadcastDS);

        //TODO 6.对关联后的数据进行处理
        SingleOutputStreamOperator<String> processDS = connectDS.process(
                new BroadcastProcessFunction<WaterSensor, Integer, String>() {

                    //处理非广播流数据
                    @Override
                    public void processElement(WaterSensor ws, BroadcastProcessFunction<WaterSensor, Integer, String>.ReadOnlyContext ctx, Collector<String> out) throws Exception {
                        //获取广播状态
                        ReadOnlyBroadcastState<String, Integer> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                        //从广播状态获取到阈值
                        Integer threshold = broadcastState.get("threshold");
                        Integer vc = ws.getVc();
                        if (vc > threshold) {
                            out.collect("当前水位值" + vc + "超出了阈值" + threshold);
                        }
                    }
                    //处理广播流数据

                    @Override
                    public void processBroadcastElement(Integer value, BroadcastProcessFunction<WaterSensor, Integer, String>.Context ctx, Collector<String> out) throws Exception {
                        //获取广播广播状态
                        BroadcastState<String, Integer> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                        //将广播流中的数据放到广播状态状态中存起来
                        broadcastState.put("threshold", value);
                    }
                }
        );
        //TODO 7.打印输出
        processDS.printToErr();
        //TODO 8.提交作业
        env.execute();
    }
}
